package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwdTrafficBaseLogSplit {
    public static void main(String[] args) {
        //基本环境准备
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);

        //检查点配置
/*        env.enableCheckpointing(5000l);
        env.getCheckpointConfig().setCheckpointTimeout(60000l);
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000l);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(10)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/
        //读取kafka的topic_log主题
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource("topic_log", "dwd_traffic_log_split_group");
        DataStreamSource<String> logDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "log");
        //过滤脏数据 , 并转换数据类型
        OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag") {};
        SingleOutputStreamOperator<JSONObject> jsonObjDs = logDs.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        try {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            out.collect(jsonObject);
                        } catch (Exception e) {
                            ctx.output(dirtyTag, jsonStr);
                        }
                    }
                }
        );
        //将脏数据写出到kafka单独主题
        jsonObjDs.getSideOutput(dirtyTag).sinkTo(
                MyKafkaUtil.getKafkaSink("dirtyTag")
        );
        jsonObjDs.print();
        //新老访客修复
        SingleOutputStreamOperator<JSONObject> isNewDs = jsonObjDs
                .keyBy(
                        value -> value.getJSONObject("common").getString("mid")
                )
                .process(
                new ProcessFunction<JSONObject, JSONObject>() {

                    private ValueState<String> userState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<String> stateProperties = new ValueStateDescriptor<String>("userState", String.class);
                        //stateProperties.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        userState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        //取时间字段
                        Long ts = jsonObj.getLong("ts");
                        JSONObject common = jsonObj.getJSONObject("common");
                        String isNew = common.getString("is_new");
                        //取状态
                        String state = userState.value();
                        if ("1".equals(isNew)) {
                            //判断状态是否为空
                            if (StringUtils.isEmpty(state)) {
                                //如果为空则为新访客,将时间写入状态
                                userState.update(DateFormatUtil.toDate(ts));
                            } else {
                                String date = DateFormatUtil.toDate(ts);
                                if (!date.equals(state)) {
                                    //如果不为空但日期与状态相同,也是新访客,不同则为,老访客
                                    common.put("is_new", "0");
                                }
                            }
                        } else {
                            //判断状态是否为空
                            if (StringUtils.isEmpty(state)) {
                                //如果为空则时间未记录,写入昨天时间
                                userState.update(DateFormatUtil.toDate(ts - 3600 * 24));
                            }
                        }
                        out.collect(jsonObj);
                    }
                }
        );
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> videoTag = new OutputTag<String>("videoTag") {};
        //日志分流
        SingleOutputStreamOperator<String> pageDS = isNewDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {

                        //根据是否有err判断错误日志
                        JSONObject err = jsonObj.getJSONObject("err");
                        if (err != null) {
                            ctx.output(errTag, jsonObj.toJSONString());
                            jsonObj.remove("err");
                        }

                        JSONObject start = jsonObj.getJSONObject("start");
                        JSONObject appVideo = jsonObj.getJSONObject("appVideo");
                        if (start != null) {
                            //根据是否有start判断启动日志
                            ctx.output(startTag, jsonObj.toJSONString());
                        } else if (appVideo != null) {
                            //根据是否有appVideo判断视频日志
                            ctx.output(videoTag, jsonObj.toJSONString());
                        } else {
                            //其余的则为页面日志
                            JSONObject common = jsonObj.getJSONObject("common");
                            JSONObject page = jsonObj.getJSONObject("page");
                            String ts = jsonObj.getString("ts");
                            //在页面日志中根据displays判断曝光日志
                            JSONArray displays = jsonObj.getJSONArray("displays");
                            if (displays != null && displays.size() > 0) {
                                for (Object display : displays) {
                                    JSONObject jsonObject = new JSONObject();
                                    jsonObject.put("display", display);
                                    jsonObject.put("common", common);
                                    jsonObject.put("page", page);
                                    jsonObject.put("ts", ts);
                                    ctx.output(displayTag, jsonObject.toJSONString());
                                }
                                jsonObj.remove("displays");
                            }
                            //在页面日志中根据actions判断动作日志
                            JSONArray actions = jsonObj.getJSONArray("actions");
                            if (actions != null && actions.size() > 0) {
                                for (Object action : actions) {
                                    JSONObject jsonObject = new JSONObject();
                                    jsonObject.put("display", action);
                                    jsonObject.put("common", common);
                                    jsonObject.put("page", page);
                                    jsonObject.put("ts", ts);
                                    ctx.output(actionTag, jsonObject.toJSONString());
                                }
                                jsonObj.remove("actions");
                            }
                            out.collect(jsonObj.toJSONString());
                        }

                    }
                }
        );
        SideOutputDataStream<String> errDs = pageDS.getSideOutput(errTag);
        SideOutputDataStream<String> startDs = pageDS.getSideOutput(startTag);
        SideOutputDataStream<String> displayDs = pageDS.getSideOutput(displayTag);
        SideOutputDataStream<String> actionDs = pageDS.getSideOutput(actionTag);
        SideOutputDataStream<String> videoDs = pageDS.getSideOutput(videoTag);
        //写入到不同主题
        pageDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_page_log"));
        startDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_start_log"));
        errDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_err_log"));
        displayDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_display_log"));
        actionDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_action_log"));
        videoDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_video_log"));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
